package cn._51doit.live.utils;

import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.kafka.clients.consumer.ConsumerRecord;

import java.util.Arrays;
import java.util.List;
import java.util.Properties;

/**
 * Flink通用的工具类
 */
public class FlinkUtils {

    public static final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


    public static <T> DataStream<T> createKafkaStreamWithId(ParameterTool parameterTool, Class<? extends KafkaDeserializationSchema<T>> clazz) throws Exception {

        int checkpointInterval = parameterTool.getInt("checkpoint.interval", 60000);
        env.enableCheckpointing(checkpointInterval);
        //env.setStateBackend(new EmbeddedRocksDBStateBackend(true)); //支持增量checkpoint
        //env.getCheckpointConfig().setCheckpointStorage("hdfs://node-1.51doit.cn:9000/ck666");
        Properties properties = new Properties();
        String bootstrapServers = parameterTool.getRequired("bootstrap.servers");
        properties.setProperty("bootstrap.servers", bootstrapServers);
        String autoOffsetReset = parameterTool.get("auto.offset.reset", "earliest");
        properties.setProperty("auto.offset.reset", autoOffsetReset);
        String enableAutoCommit = parameterTool.get("enable.auto.commit", "false");
        properties.setProperty("enable.auto.commit", enableAutoCommit);
        String kafkaInputTopics = parameterTool.getRequired("kafka.input.topics");
        List<String> topics = Arrays.asList(kafkaInputTopics.split(","));
        FlinkKafkaConsumer<T> flinkKafkaConsumer = new FlinkKafkaConsumer<T>(
                topics,
                clazz.newInstance(),
                properties
        );
        flinkKafkaConsumer.setCommitOffsetsOnCheckpoints(false);

        DataStreamSource<T> lines = env.addSource(flinkKafkaConsumer);

        return lines;

    }

    public static <T> DataStream<T> createKafkaStream(ParameterTool parameterTool, Class<? extends DeserializationSchema<T>> clazz) throws Exception {

        int checkpointInterval = parameterTool.getInt("checkpoint.interval", 60000);
        env.enableCheckpointing(checkpointInterval);
        //env.setStateBackend(new EmbeddedRocksDBStateBackend(true)); //支持增量checkpoint
        //env.getCheckpointConfig().setCheckpointStorage("hdfs://node-1.51doit.cn:9000/ck666");
        Properties properties = new Properties();
        String bootstrapServers = parameterTool.getRequired("bootstrap.servers");
        properties.setProperty("bootstrap.servers", bootstrapServers);
        String autoOffsetReset = parameterTool.get("auto.offset.reset", "earliest");
        properties.setProperty("auto.offset.reset", autoOffsetReset);
        String enableAutoCommit = parameterTool.get("enable.auto.commit", "false");
        properties.setProperty("enable.auto.commit", enableAutoCommit);
        String kafkaInputTopics = parameterTool.getRequired("kafka.input.topics");
        List<String> topics = Arrays.asList(kafkaInputTopics.split(","));
        FlinkKafkaConsumer<T> flinkKafkaConsumer = new FlinkKafkaConsumer<T>(
                topics,
                clazz.newInstance(),
                properties
        );
        flinkKafkaConsumer.setCommitOffsetsOnCheckpoints(false);

        DataStreamSource<T> lines = env.addSource(flinkKafkaConsumer);

        return lines;


    }

    public static StreamTableEnvironment getStreamTableEnv() {
        return StreamTableEnvironment.create(env);
    }
}
